package cz.data.common.elastic.manage;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import cz.data.common.core.DataConstant;
import cz.data.common.support.db.core.DbColumn;
import cz.data.common.support.db.core.DbTable;
import cz.data.common.support.metadata.EsDataTypeEnum;
import cz.data.common.utils.ThrowableUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.elasticsearch.action.ingest.GetPipelineRequest;
import org.elasticsearch.action.ingest.GetPipelineResponse;
import org.elasticsearch.action.ingest.PutPipelineRequest;
import org.elasticsearch.action.support.master.AcknowledgedResponse;
import org.elasticsearch.client.IndexLifecycleClient;
import org.elasticsearch.client.IndicesClient;
import org.elasticsearch.client.IngestClient;
import org.elasticsearch.client.Request;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.Response;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.indexlifecycle.DeleteAction;
import org.elasticsearch.client.indexlifecycle.ForceMergeAction;
import org.elasticsearch.client.indexlifecycle.LifecycleAction;
import org.elasticsearch.client.indexlifecycle.LifecyclePolicy;
import org.elasticsearch.client.indexlifecycle.Phase;
import org.elasticsearch.client.indexlifecycle.PutLifecyclePolicyRequest;
import org.elasticsearch.client.indexlifecycle.RolloverAction;
import org.elasticsearch.client.indexlifecycle.ShrinkAction;
import org.elasticsearch.client.indices.CreateIndexRequest;
import org.elasticsearch.client.indices.CreateIndexResponse;
import org.elasticsearch.client.indices.GetIndexRequest;
import org.elasticsearch.client.indices.GetMappingsRequest;
import org.elasticsearch.client.indices.GetMappingsResponse;
import org.elasticsearch.client.indices.PutMappingRequest;
import org.elasticsearch.cluster.metadata.MappingMetadata;
import org.elasticsearch.common.bytes.BytesArray;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.core.TimeValue;
import org.elasticsearch.xcontent.XContentType;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Properties;
import java.util.stream.Collectors;

@Slf4j
public class ElasticsearchActuator {
    private static final String DEFAULT_POLICY = "data_platform_index_policy";
    private static final String DEFAULT_INGEST = "data_platform_ingest_pipeline";
    private static final String TEMPLATE_PREFIX = ".data_platform_index_template_";
    private static final String DESCRIBE = DataConstant.DB_METADATA.ES_FIELD_DESCRIBE_KEY;
    private final RestHighLevelClient rest;

    public ElasticsearchActuator(RestHighLevelClient rest) {
        this.rest = rest;
    }

    public List<String> indexNames() {
        Request request = new Request("get", "/_cat/indices?format=json&expand_wildcards=open");
        JSONArray result = this.request(request, JSONArray.class);
        return result.stream()
                .map(ele -> (JSONObject) ele)
                .map(json -> json.getString("index"))
                .filter(index -> !StringUtils.startsWithIgnoreCase(index, "."))
                .collect(Collectors.toList());
    }

    public DbTable tableMapping(String index) {
        GetMappingsRequest request = new GetMappingsRequest();
        request.indices(index);
        try {
            GetMappingsResponse response = this.rest.indices().getMapping(request, RequestOptions.DEFAULT);
            Map<String, Object> mappingMap = Optional.of(response)
                    .map(GetMappingsResponse::mappings)
                    .map(mappings -> mappings.get(index))
                    .map(MappingMetadata::sourceAsMap)
                    .orElseGet(HashMap::new);
            String remark = Optional.ofNullable(mappingMap.get("_meta"))
                    .map(this::toJsonObject)
                    .map(metaJson -> metaJson.get(DESCRIBE))
                    .map(Object::toString)
                    .orElseGet(() -> "未描述表" + index);
            return DbTable.from(index, remark);
        } catch (Exception e) {
            throw new RuntimeException(e.getMessage(), e);
        }
    }

    public List<DbColumn> columnMapping(String index) {
        GetMappingsRequest request = new GetMappingsRequest();
        request.indices(index);
        try {
            GetMappingsResponse response = this.rest.indices().getMapping(request, RequestOptions.DEFAULT);
            Map<String, Object> mappingMap = Optional.of(response)
                    .map(GetMappingsResponse::mappings)
                    .map(mappings -> mappings.get(index))
                    .map(MappingMetadata::sourceAsMap)
                    .orElseGet(HashMap::new);
            JSONObject properties = Optional.of(mappingMap.get("properties"))
                    .map(this::toJsonObject)
                    .orElseGet(JSONObject::new);
            return this.columnResolver(properties);
        } catch (Exception e) {
            throw new RuntimeException(e.getMessage(), e);
        }
    }

    private List<DbColumn> columnResolver(JSONObject json) {
        List<DbColumn> columns = new ArrayList<>();
        int i = 1;
        for (Map.Entry<String, Object> entry : json.entrySet()) {
            String columnName = entry.getKey();
            JSONObject properties = Optional.of(entry.getValue())
                    .map(this::toJsonObject)
                    .orElseGet(JSONObject::new);
            String remark = Optional.ofNullable(properties.getJSONObject("meta"))
                    .map(meta -> meta.getString(DESCRIBE))
                    .orElseGet(() -> "未描述" + columnName);
            String dataType = properties.getString("type");
            DbColumn column = new DbColumn();
            column.setColName(columnName);
            column.setColComment(remark);
            column.setDataType(dataType);
            column.setColKey(Boolean.FALSE);
            column.setColPosition(i);
            column.setNullable(Boolean.TRUE);
            column.setDataDefault(properties.getString("null_value"));
            columns.add(column);
            i++;
        }
        return columns;
    }

    public <T> T request(Request request, Class<T> tClass) {
        try {
            Response response = this.rest.getLowLevelClient().performRequest(request);
            return JSONObject.parseObject(response.getEntity().getContent(), tClass);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    private JSONObject toJsonObject(Object value) {
        if (value instanceof Map) {
            JSONObject json = new JSONObject();
            Map<?, ?> map = (Map<?, ?>) value;
            map.forEach((k, v) -> json.put(k.toString(), v));
            return json;
        } else {
            return new JSONObject();
        }
    }

    public boolean createLifecyclePolicy(String policyName) {
        RestClient lowLevelClient = this.rest.getLowLevelClient();
        String endpoint = "/_ilm/policy/" + policyName;
        Boolean exists = Optional.ofNullable(
                        ThrowableUtil.callable(() -> lowLevelClient.performRequest(new Request("GET", endpoint)), false)
                ).map(Response::getStatusLine)
                .map(statusLine -> statusLine.getStatusCode() == 200)
                .orElse(false);
        if (Boolean.TRUE.equals(exists)) {
            return true;
        }
        IndexLifecycleClient client = this.rest.indexLifecycle();
        Map<String, Phase> phases = new HashMap<>();
        Map<String, LifecycleAction> warmAction = new HashMap<>();
        warmAction.put("shrink", new ShrinkAction(1, null));
        warmAction.put("forcemerge", new ForceMergeAction(1));
        phases.put("hot", new Phase("hot", null,
                        Collections.singletonMap("rollover",
                                new RolloverAction(ByteSizeValue.ofGb(10), null,
                                        null, 10000000L))
                )
        );
        phases.put("warm", new Phase("warm", TimeValue.timeValueDays(60), warmAction));
        phases.put("delete", new Phase("delete", TimeValue.timeValueDays(365),
                Collections.singletonMap("delete", new DeleteAction())));

        LifecyclePolicy policy = new LifecyclePolicy(policyName, phases);
        PutLifecyclePolicyRequest request = new PutLifecyclePolicyRequest(policy);
        return Optional.ofNullable(
                        ThrowableUtil.callable(
                                () -> client.putLifecyclePolicy(request, RequestOptions.DEFAULT), false)
                )
                .map(org.elasticsearch.client.core.AcknowledgedResponse::isAcknowledged)
                .orElse(false);
    }

    public boolean createIndex(DbTable table, Properties properties) {
        IndicesClient client = this.rest.indices();
        try {
            boolean exists = client.exists(new GetIndexRequest(table.getTableName()), RequestOptions.DEFAULT);
            if (exists) {
                PutMappingRequest request = new PutMappingRequest(table.getTableName());
                request.source(createMappings(table.getTableComment(), table.getColumns()));
                AcknowledgedResponse response = client.putMapping(request, RequestOptions.DEFAULT);
                return response.isAcknowledged();
            } else {
                CreateIndexRequest request = new CreateIndexRequest(table.getTableName());
                request.settings(defaultSettings(properties));
                request.mapping(createMappings(table.getTableComment(), table.getColumns()));
                CreateIndexResponse response = client.create(request, RequestOptions.DEFAULT);
                return response.isAcknowledged();
            }
        } catch (Exception e) {
            log.warn("索引{}创建失败: {}", table.getTableName(), e.getMessage());
            return false;
        }
    }

    public boolean createIndexTemplate(DbTable table, Properties properties) {
        JSONObject payload = new JSONObject();
        JSONObject template = new JSONObject();
        JSONObject settings = defaultSettings(properties);
        settings.put("index.lifecycle.name", DEFAULT_POLICY);
        settings.put("index.lifecycle.rollover_alias", table.getTableName());
        settings.put("index.default_pipeline", DEFAULT_INGEST);
        template.put("settings", settings);
        template.put("mappings", createMappings(table.getTableComment(), table.getColumns()));
        if (properties.containsKey("data_stream")) {
            String streamName = properties.getProperty("data_stream");
            payload.put("index_patterns", Collections.singletonList(streamName + "*"));
            payload.put("data_stream", new JSONObject());
        } else {
            payload.put("index_patterns", Collections.singletonList(table.getTableName() + "-*"));
            template.put("aliases", new JSONObject().fluentPut(table.getTableName(), new JSONObject()));
        }
        payload.put("template", template);
        payload.put("priority", 100);
        payload.put("_meta", new JSONObject().fluentPut(DESCRIBE, table.getTableName() + "索引模版"));
        String templateName = TEMPLATE_PREFIX + table.getTableName();
        Request request = new Request("PUT", "/_index_template/" + templateName);
        request.setJsonEntity(payload.toJSONString());
        RestClient lowLevelClient = this.rest.getLowLevelClient();
        log.info(payload.toJSONString());
        return Optional.ofNullable(
                        ThrowableUtil.callable(
                                () -> lowLevelClient.performRequest(request)
                        )
                ).map(Response::getStatusLine)
                .map(statusLine -> statusLine.getStatusCode() == 200)
                .orElse(false);
    }

    public boolean createDefaultIngestPipeline() {
        IngestClient client = this.rest.ingest();
        Boolean exists = Optional.ofNullable(
                        ThrowableUtil.callable(
                                () -> client.getPipeline(new GetPipelineRequest(DEFAULT_INGEST), RequestOptions.DEFAULT),
                                false)
                ).map(GetPipelineResponse::isFound)
                .orElse(false);
        if (exists) {
            return true;
        }
        JSONObject script = new JSONObject();
        script.put("source", "if (ctx['@timestamp'] == null) {\n" +
                "ctx['@timestamp'] = new Date();\n}");
        JSONObject payload = new JSONObject();
        payload.put("description", "数据开发平台默认管道,增加一个@timestamp字段");
        payload.put("processors", Collections.singletonList(new JSONObject().fluentPut("script", script)));
        return Optional.ofNullable(
                        ThrowableUtil.callable(() ->
                                client.putPipeline(new PutPipelineRequest(DEFAULT_INGEST,
                                                new BytesArray(payload.toJSONString()), XContentType.JSON),
                                        RequestOptions.DEFAULT), false
                        )
                ).map(AcknowledgedResponse::isAcknowledged)
                .orElse(false);
    }

    public JSONObject defaultSettings(Properties properties) {
        String numberOfShards = properties.getProperty("number_of_shards", "3");
        String numberOfReplicas = properties.getProperty("number_of_replicas", "0");
        JSONObject settings = new JSONObject();
        settings.put("index.number_of_shards", numberOfShards);
        settings.put("index.number_of_replicas", numberOfReplicas);
        settings.put("index.refresh_interval", "10s");
        return settings;
    }

    public JSONObject createMappings(String tableComment, List<DbColumn> columns) {
        JSONObject mappings = new JSONObject();
        mappings.put("dynamic", "false");
        mappings.put("_meta", new JSONObject().fluentPut(DESCRIBE, tableComment));
        JSONObject properties = new JSONObject();
        for (DbColumn column : columns) {
            JSONObject field = new JSONObject();
            field.put("type", column.getDataType());
            if (EsDataTypeEnum.DATE.equalsValue(column.getDataType())) {
                field.put("format", "epoch_millis||yyyy-MM-dd||yyyy-MM-dd HH:mm:ss||yyyyMMddHHmmss");
            } else if (EsDataTypeEnum.KEYWORD.equalsValue(column.getDataType())
                    || EsDataTypeEnum.TEXT.equalsValue(column.getDataType())) {
                field.put("copy_to", "_search_text");
            }
            field.put("meta", new JSONObject().fluentPut(DESCRIBE, column.getColComment()));
            properties.put(column.getColName(), field);
        }
        properties.put("_search_text", new JSONObject().fluentPut("type", "text")
                .fluentPut("meta", new JSONObject().fluentPut(DESCRIBE, "全文检索字段")));
        mappings.put("properties", properties);
        return mappings;
    }
}
